package com.greenline.push.service.core.task;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.greenline.push.core.enums.TaskStatus;
import com.greenline.push.core.enums.Tasks;
import com.greenline.push.core.pushcenter.MsgForSendManager;
import com.greenline.push.core.pushcenter.MsgHisManager;
import com.greenline.push.core.pushcenter.TaskSyncManager;
import com.greenline.push.dal.pushcenter.dataobject.MsgForSendDO;
import com.greenline.push.dal.pushcenter.dataobject.TaskSyncDO;
import com.greenline.push.service.core.ProcessQueueList;
import com.greenline.push.service.core.exception.TaskException;
import com.greenline.push.service.core.item.Packet;
import com.greenline.push.service.core.utils.Constants;
import com.greenline.push.service.core.utils.PacketConvertor;

/**
 * 
 * 定时任务取数据
 * 
 * @author zhangdp
 * 
 */
public class FetchFixedTimeDataAsyncSchedule extends TaskBase implements Runnable {

    private ProcessQueueList processQueueList;

    private final Integer MAX_MSG_COUNT = 1000;

    private TaskSyncDO task = null;
    private final String taskName = "{获取待发送任务(定时)}";
    private Tasks taskType = null;
    @Autowired
    private TaskSyncManager taskSyncManager;

    @Autowired
    private MsgForSendManager msgForSendManager;

    @Autowired
    private MsgHisManager msgHisManager;

    @Autowired
    private PacketConvertor packetConvertor;
    private Logger log = Constants.LOG_SCHEDULE;
    // 定时任务没执行一次，生成唯一的识别号
    private String taskId = null;

    public FetchFixedTimeDataAsyncSchedule(ProcessQueueList queue) {
        this.processQueueList = queue;
    }

    public void startTask(ProcessQueueList processQueueList, Tasks task, long period, TimeUnit unit) throws TaskException {
        log.debug(String.format("%s--任务类型:<%s>开始启动----------", taskName, task));
        if (processQueueList == null || task == null) {
            throw new TaskException("任务启动出错,queue不可为空,task不可为空");
        }
        // task init from here
        ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1);
        this.processQueueList = processQueueList;
        this.taskType = task;
        schedule.scheduleAtFixedRate(this, 0, period, unit);
        log.debug(String.format("taskName 任务类型:<%s>启动完成,运行周期:<%s>=========", task, period));
    }

    public FetchFixedTimeDataAsyncSchedule(ProcessQueueList processQueueList, Tasks taskType) {
        this.processQueueList = processQueueList;
        this.taskType = taskType;
    }

    public FetchFixedTimeDataAsyncSchedule() {

    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        excuteTask(new TaskCallBack() {
            /**
             * check task whether start
             */
            @Override
            public TaskInfo initTaskInfo() {
                task = taskSyncManager.getTaskSyncByType(taskType.getType());
                // get uuid
                taskId = UUID.randomUUID().toString();

                log.debug(String.format("------定时任务开始执行----taskType:%s,tasks:%s,taskId:%s", taskType.getType(), taskType, taskId));

                TaskInfo taskInfo = new TaskInfo(taskName);
                // 无任务或者任务状态为running（status=1为running），task不能启动
                if (task == null) {
                    // here ,this return will broken the task forever.
                    log.debug(taskType + "-没有找到任务,停止执行。" + taskId);
                    return null;
                } else if (task.getStatus() == TaskStatus.RUNNING.getStatus()) {
                    log.debug(taskType + "-任务碰撞，停止执行。" + taskId);
                    taskInfo.setBizResult(TaskStatus.RUNNING.getName());
                    return taskInfo;
                } else {
                    task.setStatus(TaskStatus.RUNNING.getStatus());
                    taskSyncManager.update(task);

                    taskInfo.setBizResult(TaskStatus.END.getName());
                    return taskInfo;
                }
            }

            @Override
            public TaskInfo excute(TaskInfo taskInfo) {

                if (task != null) {
                    Long maxMsgId = task.getLastRecordId();
                    List<MsgForSendDO> msgList = msgForSendManager.getPeriodicMsgForSendList(maxMsgId, MAX_MSG_COUNT);
                    if (msgList != null) {
                        int size = msgList.size();

                        // 保存上次入队的最大的msg id
                        Long lastMaxMsgId = task.getLastRecordId();
                        // flag和count用来标识在入队列的时候发生错误
                        boolean flag = true;
                        int count = 0;
                        // 如果process发生异常，任务失败退出。
                        boolean taskFlag = true;
                        // 用于保存maxMsgId的中间值
                        long tempId = 0l;
                        try {
                            // 先排序，确保id从小到大
                            MsgComparator<MsgForSendDO> c = new MsgComparator<MsgForSendDO>();
                            Collections.sort(msgList, c);
                            for (int i = 0; i < size; i++) {
                                MsgForSendDO msg = msgList.get(i);
                                tempId = msg.getId();
                                Packet packet = packetConvertor.convertForSendDOToPacket(msg);

                                boolean queueFlag = processQueueList.offer(packet);
                                if (!queueFlag) {
                                    flag = false;
                                    count++;
                                }
                            }
                        } catch (Exception e) {

                            taskFlag = false;
                            log.error(taskType + "执行异常:" + e.getMessage() + "任务id:" + taskId);
                            // then throw exception
                        } finally {
                            // if flag is true and count == 0, all the msg
                            // pushed to queue successfully
                            if (flag && count == 0) {
                                if (size != 0) {
                                    lastMaxMsgId = msgList.get(size - 1).getId();
                                }
                            } else if (!flag && count == 1) {
                                lastMaxMsgId = tempId;
                            }
                            task.setType(taskType.getType());
                            task.setLastRecordId(lastMaxMsgId);
                            // set status as end
                            task.setStatus(TaskStatus.END.getStatus());
                            taskSyncManager.update(task);
                            if (taskFlag) {

                                taskInfo.setSeccusss(true);
                                taskInfo.setBizResult("成功");
                            } else {
                                // task_info 处理
                                task.setStatus(TaskStatus.END.getStatus());
                                taskSyncManager.update(task);
                                taskInfo.setSeccusss(false);
                                taskInfo.setBizResult("失败");
                            }

                        }

                    } else {
                        System.err.println("没有数据需要拉取");
                        task.setStatus(TaskStatus.END.getStatus());
                        taskSyncManager.update(task);
                        taskInfo.setSeccusss(false);
                        taskInfo.setBizResult("失败");
                    }
                } else {
                    System.err.println("没有相应的task配置");
                    task.setStatus(TaskStatus.END.getStatus());
                    taskSyncManager.update(task);
                    taskInfo.setSeccusss(false);
                    taskInfo.setBizResult("失败");
                }
                log.debug(String.format("%s--%s:--任务执行结束***任务id:%s", taskName, taskType, taskId));
                return taskInfo;
            }
        });
    }

    private class MsgComparator<T> implements Comparator<T> {

        @Override
        public int compare(T o1, T o2) {
            MsgForSendDO msg1 = (MsgForSendDO) o1;
            MsgForSendDO msg2 = (MsgForSendDO) o2;
            long id1 = msg1.getId();
            long id2 = msg2.getId();
            if (id1 > id2) {
                return 1;
            } else if (id1 < id2) {
                return -1;
            } else {
                return 0;
            }
        }

    }
}
